package com.zhao.biz.report.report2_region.version2_core

import com.zhao.entity.Log
import com.zhao.utils.ReportUtils
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession

/**
 * Description: 报表展示之地域分布 spark core版本<br/>
 * Copyright (c) ，2021 ， 赵 <br/>
 * A wet person does not fear the rain. <br/>
 * Date： 2021/1/13 16:16
 *
 * @author 柒柒
 * @version : 1.0
 */

object RegionCal {
  def main(args: Array[String]): Unit = {
    //步骤:
    val inputPath = "a_data/outputpath"
    val Array(input) = Array(inputPath)

    //1.SparkSession
    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName)
      .master("local[*]")
      //设置序列化的技术(使用kryo)
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .getOrCreate()

    val sc: SparkContext = spark.sparkContext

    //2.注册实体类(那个实体类的实例使用到了kryo序列化技术)
    sc.getConf.registerKryoClasses(Array(classOf[Log]))

    spark.read.parquet(input).rdd
      .map(row =>{
        //先把需要的字段拿出来,在进行请求
        //处理原始请求数,有效请求,广告请求
        //requestmode 数据请求方式(1:请求,2:展示,3:点击)
        val requestmode = row.getAs[Int]("requestmode")
        //procenode 流程节点(1.请求量kpi 2.有效请求 3.广告请求)
        val processnode = row.getAs[Int]("processnode")
        //iseffective 有效标识(有效值可以正常计费的)(0:无效 1:有效)
        val iseffective = row.getAs[Int]("iseffective")
        //isbilling 是否收费(0:未收费 1:已收费)
        val isbilling = row.getAs[Int]("isbilling")
        //isbid 是否rtb
        val isbid = row.getAs[Int]("isbid")
        //iswin 是否竞价成功
        val iswin = row.getAs[Int]("iswin")
        //adorderid 广告id
        val adorderid = row.getAs[Int]("adorderid")
        //adpayment 转换后广告消费
        val adpayment = row.getAs[Double]("adpayment")
        //winprice rtb竞争成功价格
        val winPrice = row.getAs[Double]("winprice")

        //设置工具方法进行计算,计算的结果是List
        //处理 原始请求,有效请求广告请求
        val reqList: List[Double] = ReportUtils.calculateReq(requestmode, processnode)
        //参与竞价数
        val rtbCntList: List[Double] = ReportUtils.calculateRtbCnt(iseffective, isbilling, isbid)
        //竞价成功数,广告消费,广告成本
        val rtbList = ReportUtils.calculateRtb(iseffective, isbilling, isbid, iswin, adorderid, winPrice, adpayment)
        //展示数,点击数
        val clickList: List[Double] = ReportUtils.calculateTimes(requestmode, iseffective)

        //聚合的key
        val provinceName = row.getAs[String]("provincename")
        val cityName = row.getAs[String]("cityname")
        ((provinceName,cityName),reqList ++ rtbCntList ++ rtbList ++ clickList)
      }).reduceByKey((lst1,lst2) =>{
      lst1.zip(lst2)
        .map(perEle => perEle._1 + perEle._2)
    }).map(perEle =>{
      val provinceName = perEle._1._1
      val cityName = perEle._1._2

      val lst: List[Double] = perEle._2

      //处理原始请求数,有效请求,广告请求
      val originalReq = lst(0)
      val effectiveReq = lst(1)
      val adReq = lst(2)

      //参与竞价数,竞价成功数,广告消费,广告成本
      val bidCnt = lst(3)
      val bidSuccCnt = lst(4)
      val adConsumer = lst(5)
      val adCost = lst(6)

      //展示数,点击数
      val showCnt = lst(7)
      val clickCnt = lst(8)

      (provinceName,
      cityName,
      originalReq,
      effectiveReq,
      adReq,
      bidCnt,
      bidSuccCnt,
      if (bidCnt == 0) 0 else (bidSuccCnt * 100.0/bidCnt).formatted("%.2f"),
      showCnt,
      clickCnt,
      if (showCnt == 0) 0 else (clickCnt * 100.0/showCnt).formatted("%.2f"),
      adCost,
      adConsumer)
    }).collect.foreach(println)

    //资源释放
    spark.stop()
  }
}


















